package tools;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;

public class CatchFileTools {

	public static Job addCache(Job job,Configuration conf,String cachePath,String cacheMark) throws IOException{
		FileSystem hdfs = FileSystem.get(conf);
		FileStatus[] fileList = hdfs.listStatus(new Path(cachePath));
		for (int i = 0; i < fileList.length; i++)
		{
			String fileName = fileList[i].getPath().getName();
			String fullFilePath=cachePath+"/"+fileName;
			String pathLink = new Path(fullFilePath).toUri().toString()+"#"+cacheMark+i;

			try
			{
				DistributedCache.addCacheFile(new URI(pathLink), job.getConfiguration());
			}
			catch (URISyntaxException e)
			{
				e.printStackTrace();
			}
		}
		hdfs.close();
		return job;
	}

	public static List<String> getListByCache(Path[] paths,String pathMark,String charSet){
		List<String> list = new ArrayList<String>();
		BufferedReader in = null;
		try {
			String line = null;
			for (Path path : paths) {
				if (path.toString().contains(pathMark)) {
					in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()), charSet)));
					while (null != (line = in.readLine())) {
						list.add(line);
					}
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try { if (in != null) { in.close(); } } catch (IOException e) {}
		}
		return list;
	}

	public static String getSubArrStr(String [] strArr,int [] indexs,String sep){
		String string="";
		if(indexs==null){
			return string;
		}

		for(int i=0;i<indexs.length;i++){
			String tmp="";
			try{tmp=strArr[indexs[i]];}catch(Exception e){tmp="";}
			if(i!=0){string+=sep;}
			string+=tmp;
		}
		return string;
	}

	public static Map<String,String> getMapByCache(Path[] paths,String pathMark,String charSet,String sep,int [] keyIndexs,int [] valIndexs){
		Map<String,String> map = new HashMap<String,String>();
		BufferedReader in = null;
		try {
			String line = null;
			for (Path path : paths) {
				if(!path.toString().contains(pathMark)){
					continue;
				}
				in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()), charSet)));
				while (null != (line = in.readLine())) {
					String[] array = line.split(sep, -1);
					String strKey=getSubArrStr(array, keyIndexs, sep);
					String strVal=getSubArrStr(array, valIndexs, sep);
					map.put(strKey, strVal);
				}
			}
		} catch (IOException e) {
			//e.printStackTrace();
		} finally {
			try {
				if (in != null) {
					in.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return map;
	}

	public static Map<String, String> getMapByCache(Path[] paths,String pathMark, String charSet, String sep) {
		Map<String, String> map = new HashMap<String, String>();
		BufferedReader in = null;
		try {
			String line = null;
			for (Path path : paths) {
				if (path.toString().contains(pathMark)) {
					in = new BufferedReader((new InputStreamReader(new FileInputStream(path.toString()), charSet)));
					while (null != (line = in.readLine())) {
						String[] array = line.split(sep, 2);
						if (array.length != 2) {
							continue;
						} else {
							map.put(array[0], array[1]);
						}
					}
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				if (in != null) {
					in.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return map;
	}


	public static Map<String, String> getCityCateIdMapByCache(Path[] paths,
			String pathMark, String charSet, String sep,String flag) {
		Map<String, String> map = new HashMap<String, String>();
		BufferedReader in = null;
		try {
			String line = null;
			for (Path path : paths) {
				if (path.toString().contains(pathMark)) {
					in = new BufferedReader((new InputStreamReader(
							new FileInputStream(path.toString()), charSet)));
					while (null != (line = in.readLine())) {
						String [] strArr=line.split(sep,-1);
						List<String> list = new ArrayList<String>();
						for(int i=0;i<strArr.length;i++){
							String s1 = strArr[i];
							if(s1==null||s1.equals("null")||s1.equals("NULL")||s1.equals("")||s1.equals("\\N")){
								s1="";
							}
							if(StringUtils.isNotBlank(s1)){
								list.add(s1);
							}

						}
						String newKey = "";
						String newVal = "";
						if(flag.equals("city") && !list.get(0).equals("8728")){
							newKey = list.get(0);
							newVal = list.get(2);
						}else if(flag.equals("cate")){
							newKey = list.get(0);
							newVal = list.get(2);
						}else if(flag.equals("bussiness")){
							newKey = list.get(2);
							newVal = list.get(0);
						}else if(flag.equals("cateidPath")){
							newKey = list.get(0);
							newVal = list.get(5);
						}else if(flag.equals("cityidPath") && !list.get(0).equals("8728")){
							//8728 表示全国,要过滤
							newKey = list.get(0);
							newVal = list.get(4);
						}

						map.put(newKey,newVal);
					}
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				if (in != null) {
					in.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return map;
	}
	// 1,11 4
	// 根据id全路径取出对应的名称全路径
//	public static Map<String, String> getIdToNamePath(Map<String,String> idPathMap,Map<String,String> idToNameMap,int len) {
//		Map<String, String> namePathMap = new HashMap<String, String>();
//		//将id=0的情况加进去
//		String name_0 = "";
//		for (int i = 0; i < len; i++) {
//			name_0 += "all\t";
//		}
//		namePathMap.put("0", name_0.substring(0, name_0.length()-1));
//		namePathMap.put("-101", name_0.substring(0, name_0.length()-1));
//		namePathMap.put("-102", name_0.substring(0, name_0.length()-1));
//		//下面的是正常的id
//		String name = "";
//		for (Map.Entry<String, String> entry : idPathMap.entrySet()) {
//			String id  = entry.getKey();
//			String[] arrys = entry.getValue().split(",");
//			int j = arrys.length>=len?len:arrys.length;
//			for (int i = 0; i < j; i++) {
//				if(idToNameMap.get(arrys[i])!=null){
//					name+=idToNameMap.get(arrys[i])+"\t";
//				}else{
//					break;
//				}
//			}
//
//			int a = len;
//			if(StringUtils.isNotBlank(name)){
//				a = len-name.split("\t").length;
//			}
//			for (int i = 0; i < a; i++) {
//				name+="all\t";
//			}
//			namePathMap.put(id, name.substring(0, name.length()-1));
//		}
//		return namePathMap;
//	}
	public static void main(String args[]){
		String a =  "1,11,111,1111,";
		System.out.println(a.split(",").length);
	}
}
